package com.atguigu.flink.day09;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @author Felix
 * @date 2023/12/11
 * 该案例演示了检查点相关的设置
 */
public class Flink01_checkpoint {
    public static void main(String[] args) throws Exception {
        //执行流处理环境
        // Configuration conf = new Configuration();
        // StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度
        env.setParallelism(1);

        // 启用检查点  interval:检查点周期
        // CheckpointingMode.EXACTLY_ONCE(默认值)---barrier对齐的精准一次
        // CheckpointingMode.AT_LEAST_ONCE      ---barrier对齐的至少一次
        env.enableCheckpointing(5000L,CheckpointingMode.EXACTLY_ONCE);
        CheckpointConfig checkpointConfig = env.getCheckpointConfig();

        // 检查点存储位置  默认存在jobmanager的堆内存
        // checkpointConfig.setCheckpointStorage(new JobManagerCheckpointStorage());
        checkpointConfig.setCheckpointStorage("hdfs://hadoop102:8020/ck");

        // 检查点模式（CheckpointingMode）  也可以在开启检查点的时候指定
        // checkpointConfig.setCheckpointingMode(CheckpointingMode.AT_LEAST_ONCE);

        // 超时时间（checkpointTimeout）
        checkpointConfig.setCheckpointTimeout(60000L);

        // 最小间隔时间（minPauseBetweenCheckpoints）  上一个检查点结束和下一个检查点起始时间间隔
        //如果设置了最小时间间隔，那么最大并发检查点数据只能为1
        checkpointConfig.setMinPauseBetweenCheckpoints(2000L);

        // 最大并发检查点数量（maxConcurrentCheckpoints）   可以同时有多少个检查点在进行备份
        // checkpointConfig.setMaxConcurrentCheckpoints(1);

        // 开启外部持久化存储（enableExternalizedCheckpoints）--job取消后，检查点是否保留
        checkpointConfig.setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // checkpointConfig.setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION);

        // 检查点连续失败次数（tolerableCheckpointFailureNumber）
        checkpointConfig.setTolerableCheckpointFailureNumber(3);

        //设置操作hadoop的用户
        System.setProperty("HADOOP_USER_NAME","atguigu");

        // 设置重启策略(和容错有关)
        //flink本身提供了容错机制，如果程序出现异常，会尝试进行恢复，通过重启恢复，默认重启Integer.MAX_VALUE次
        // env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,3000L));

        // 设置状态后端
        // env.setStateBackend(new HashMapStateBackend());  ---状态存在TM堆内存
        // env.setStateBackend(new EmbeddedRocksDBStateBackend());  ---状态存在RocksDS


        // 非对齐检查点（enableUnalignedCheckpoints）
        // checkpointConfig.enableUnalignedCheckpoints();
        // 对齐检查点超时时间（alignedCheckpointTimeout）
        // checkpointConfig.setAlignedCheckpointTimeout(Duration.ofSeconds(30));

        // 通用增量 checkpoint (changelog)
        // env.enableChangelogStateBackend(true);

        // 最终检查点
        // conf.set(ExecutionCheckpointingOptions.ENABLE_CHECKPOINTS_AFTER_TASKS_FINISH, false);


        env
            .socketTextStream("hadoop102", 8888).uid("source-id")
            .flatMap(
                (String lineStr, Collector<Tuple2<String, Long>> out) ->{
                    String[] wordArr = lineStr.split(" ");
                    for (String word : wordArr) {
                        out.collect(Tuple2.of(word, 1L));
                    }
                }
                // ).returns(new TypeHint<Tuple2<String, Long>>() {})
            ).returns(Types.TUPLE(Types.STRING,Types.LONG)).uid("flatMap-id")
            .keyBy(wordTuple->wordTuple.f0)
            .sum(1).uid("sum-id")
            .print().uid("print-id");
        env.execute();
    }
}
